package com.geek.gkpush.framework.retry.core;

import com.geek.gkpush.framework.dispatch.SpringCtxUtils;
import com.geek.gkpush.framework.retry.RetryMsgHandler;
import com.geek.gkpush.framework.retry.RetryMsgPersistFacade;
import com.geek.gkpush.framework.retry.action.RetryMsgUpdateAction;
import com.geek.gkpush.framework.retry.enums.ExecStatusEnum;
import com.geek.gkpush.framework.retry.enums.RetryResultEnum;
import com.geek.gkpush.framework.retry.model.*;
import com.geek.gkpush.framework.retry.queue.RetryDelayQueue;
import com.geek.gkpush.util.excp.ExcpStackUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * <h1>重试队列处理器。</h1><br/>
 *
 * <h1><br/><a href="https://u.geekbang.org/subject/arch3/100614201">Java 业务架构实战营：https://u.geekbang.org/subject/arch3/100614201</a></h1><br/><h1><a href="https://gitee.com/ylimhhmily/BizArchPracticeCamp">Java 业务架构实战营代码地址：https://gitee.com/ylimhhmily/BizArchPracticeCamp</a></h1><br/><h1><a href="https://u.geekbang.org/subject/java4th/1001148?source=app_share">高级 Java 工程体系课 2023 版：https://u.geekbang.org/subject/java4th/1001148?source=app_share</a></h1><br/><h1><a href="https://time.geekbang.org/column/intro/100312101?tab=catalog">Dubbo 源码剖析与实战：https://time.geekbang.org/column/intro/100312101?tab=catalog</a></h1>
 *
 * @author hmilyylimh
 * ^_^
 * @version 0.0.1
 * ^_^
 * @date 2023/10/3 12:00
 */
@Slf4j
@Component
public class RetryDelayProcessor {

    @Autowired
    private RetryMsgUpdateAction retryMsgUpdateAction;
    @Autowired
    private RetryMsgPersistFacade retryMsgPersistFacade;

    private RetryDelayQueue delayQueue = null;
    private RetryMsgRunnable retryMsgRunnable = null;
    private Thread retryMsgThread = null;

    private AtomicBoolean once = new AtomicBoolean(false);

    public RetryDelayProcessor() {
        delayQueue = new RetryDelayQueue(8192);
    }

    @EventListener(value = ContextRefreshedEvent.class)
    public void refresh(ContextRefreshedEvent event) {
        if (once.compareAndSet(false, true)) {
            start();
        }
    }

    /**
     * <h2>首次添加重试任务。</h2>
     *
     * @param: retryInsertModel
     * @return: void
     */
    public void add(RetryInsertModel retryInsertModel) {
        // 1. 保存重试消息表
        RetryConfigModel retryConfig = retryMsgPersistFacade.getRetryConfig(retryInsertModel.getBizType());
        RetryMsgDTO retryMsgDTO = retryInsertModel.convert2RetryMsgDTO(retryConfig);
        retryMsgUpdateAction.saveRetryMsg(retryMsgDTO);

        // 2. 异步执行
        RetryDelayModel retryDelayModel = retryMsgDTO.createRetryDelayModel();
        delayQueue.offer(retryDelayModel);
    }

    /**
     * <h2>再次重新添加重试任务，虽然和 add 方法一摸一样，后续考虑可以在 reAdd 做些其他业务逻辑。</h2>
     *
     * @param: retryDelayModel
     * @return: void
     */
    public void reAdd(RetryDelayModel retryDelayModel) {
        if (retryDelayModel.overTimesFlag()) {
            log.warn("reAdd OverTimes, retryDelayModel={}", retryDelayModel);
            return;
        }

        delayQueue.offer(retryDelayModel);
    }

    /**
     * <h2>消费延迟队列中的数据。</h2>
     *
     * @param: task
     * @return: com.geek.gkpush.framework.retry.model.RetryResult
     */
    public RetryResult consumeDelayQueue(RetryDelayModel task) {
        try {
            return consumeDelayQueueInner(task);
        } catch (Exception e) {
            log.error("consumeDelayQueue error.");
            return RetryResult.createException(ExcpStackUtils.getStackTrace(e));
        }
    }

    /**
     * <h2>消费延迟队列中的数据。</h2>
     *
     * @param: task
     * @return: com.geek.gkpush.framework.retry.model.RetryResult
     */
    private RetryResult consumeDelayQueueInner(RetryDelayModel task) {
        // 1. 不存在，则直接结束流程
        String bizType = task.getBizType();
        RetryMsgDTO dbRetryMsgDTO = retryMsgUpdateAction.queryRetryMsg(task.getBizValue(), bizType);
        if (dbRetryMsgDTO == null) {
            return RetryResult.create(RetryResultEnum.NOT_EXISTS, task.getExecStatus());
        }
        // 2. 成功结果
        if (ExecStatusEnum.isSucc(dbRetryMsgDTO.getExecStatus())) {
            return RetryResult.create(RetryResultEnum.SUCC, task.getExecStatus());
        }
        // 3. 执行中状态
        if (ExecStatusEnum.isRunning(dbRetryMsgDTO.getExecStatus())) {
            return RetryResult.create(RetryResultEnum.LOCK_FAIL, task.getExecStatus());
        }
        // 4. 存在的话，则看看是否存在并发情况，看次数是否能对的上，对不上则说明有其他地方先行一步执行了
        if (dbRetryMsgDTO.getExecTimes() != task.getExecTimes()
                || dbRetryMsgDTO.getExecTimes() >= dbRetryMsgDTO.getMaxRetryTimes()
                || !dbRetryMsgDTO.getId().equals(task.getId())) {
            return RetryResult.create(RetryResultEnum.LOCK_FAIL, task.getExecStatus());
        }
        // 5. 尝试加锁
        boolean succ = retryMsgUpdateAction.tryLock(task);
        if (!succ) {
            return RetryResult.create(RetryResultEnum.LOCK_FAIL, task.getExecStatus());
        }
        // 6. 执行业务逻辑
        RetryMsgHandler retryMsgHandler = SpringCtxUtils.getBean(bizType, RetryMsgHandler.class);
        return retryMsgHandler.asyncProcessRetryMsg(task);
    }

    private void start() {
        // 启动线程
        retryMsgRunnable = new RetryMsgRunnable(delayQueue, this, retryMsgUpdateAction);
        retryMsgRunnable.start();

        // 设置为守护线程，JVM 退出时会杀死进程中所有的守护线程
        retryMsgThread = new Thread(retryMsgRunnable);
        retryMsgThread.setDaemon(true);
        retryMsgThread.start();

        // 添加钩子回调，便于将来释放线程
        Runtime.getRuntime().addShutdownHook(new Thread(() -> retryMsgRunnable.stop()));
    }
}